package com.zhss.seckill.inventory;

import com.alibaba.fastjson.JSONObject;
import com.zhss.seckill.common.RedisCluster;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
public class BootListener implements CommandLineRunner {

    public void run(String... strings) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(
                "seckill-inventory-consumer-group");
        consumer.setNamesrvAddr("localhost:9876");

        consumer.subscribe("seckill_product_added_topic", "*");

        consumer.registerMessageListener(new MessageListenerConcurrently() {

            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messageExts,
                                                            ConsumeConcurrentlyContext context) {
                for(MessageExt messageExt : messageExts) {
                    System.out.println(new String(messageExt.getBody()));

                    // 得知，秒杀场次里增加的是哪个商品
                    String seckillProductJSON = new String(messageExt.getBody());
                    JSONObject seckillProductJSONObject = JSONObject.parseObject(seckillProductJSON);

                    // 调用库存中心提供的接口，冻结商品用于秒杀活动的库存
                    Long productId = seckillProductJSONObject.getLong("productId");
                    Double seckillPrice = seckillProductJSONObject.getDouble("seckillPrice");
                    Long seckillStock = seckillProductJSONObject.getLong("seckillStock");

                    System.out.println("调用库存中心提供的接口，冻结商品用于秒杀活动的库存");
                    // 库存中心：可售库存、锁定库存、已售库存、冻结库存

                    // 把秒杀商品的库存进行分片，放在redis各个服务器上去
                    RedisCluster redisCluster = RedisCluster.getInstance();
                    redisCluster.initSeckillProductStock(productId,seckillStock);
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
        System.out.println("消费者启动......");
    }

}
